
package org.example.datastream.test;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.example.datastream.test.custom.DbConf;
import org.example.datastream.test.custom.Enterprise;
import org.example.datastream.test.custom.EnterpriseJdbcMultiSource;

import java.util.ArrayList;

/**
 *
 */
public class Job2 {

	public static void main(String[] args) throws Exception {
//		String limit = "limit 10";
//		if (null != args && args.length > 0) {
//			limit = " limit " + args[0];
//		}

		// set up the streaming execution environment
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		String host = "localhost";
		String port = "3306";
		String username = "root";
		String password = "root";
		String db = "test";

		ArrayList<DbConf> dbs = new ArrayList<>();
		dbs.add(new DbConf(host,port,username,password,db));
		env.addSource(new EnterpriseJdbcMultiSource(
				"mysql",
				"com.mysql.jdbc.Driver",
				dbs,
				"t_enterprise",
				2,
				1000
			))
			.name("t_enterprise_x")
			.addSink(JdbcSink.sink(
			"insert into t_enterprise(id,eid,enterprise_name,standard_name,group_name," +
					"group_abbreviation,logo_url,oper_name,oper_type,oper_name_id,title," +
					"title_code,reg_capi,rec_capi,currency_unit,business_url,collegues_num," +
					"belong_org,department,scope,reg_no,credit_no,org_no,econ_kind_code,type," +
					"type_new,category_new,status,status_code,start_date,term_start,term_end," +
					"check_date,logout_date,logout_reason,revoke_date,revoke_reason," +
					"revoked_certificates,province_code,source,created_time,row_update_time) " +
					"values(null,?,?,?,?,?,?,?,?,?,?, ?,?,?,?,?,?,?,?,?,?, ?,?,?,?,?,?,?,?,?,?, ?,?,?,?,?,?,?,?,?,?, ?)" +
					" ON DUPLICATE KEY UPDATE eid=?, enterprise_name=?, standard_name=?, group_name=?," +
					"group_abbreviation=?, logo_url=?, oper_name=?, oper_type=?, oper_name_id=?, title=?," +
					"title_code=?, reg_capi=?, rec_capi=?, currency_unit=?, business_url=?, collegues_num=?," +
					"belong_org=?,department=?,scope=?,reg_no=?,credit_no=?,org_no=?,econ_kind_code=?,type=?," +
					"type_new=?,category_new=?,status=?,status_code=?,start_date=?,term_start=?,term_end=?," +
					"check_date=?,logout_date=?,logout_reason=?,revoke_date=?,revoke_reason=?," +
					"revoked_certificates=?,province_code=?,source=?,created_time=?,row_update_time=?",
				(JdbcStatementBuilder<Enterprise>) (ps, ent) -> {
					ps.setString(1,ent.getEid());
					ps.setString(2,ent.getEnterpriseName());
					ps.setString(3,ent.getStandardName());
					ps.setString(4,ent.getGroupName());
					ps.setString(5,ent.getGroupAbbreviation());
					ps.setString(6,ent.getLogoUrl());
					ps.setString(7,ent.getOperName());
					ps.setString(8,ent.getOperType());
					ps.setString(9,ent.getOperNameId());
					ps.setString(10,ent.getTitle());

					ps.setInt(11,ent.getTitleCode());
					ps.setBigDecimal(12,ent.getRegCapi());
					ps.setBigDecimal(13,ent.getRecCapi());
					ps.setString(14,ent.getCurrencyUnit());
					ps.setString(15,ent.getBusinessUrl());
					ps.setString(16,ent.getColleguesNum());
					ps.setString(17,ent.getBelongOrg());
					ps.setString(18,ent.getDepartment());
					ps.setString(19,ent.getScope());
					ps.setString(20,ent.getRegNo());

					ps.setString(21,ent.getCreditNo());
					ps.setString(22,ent.getOrgNo());
					ps.setInt(23,ent.getEconKindCode());
					ps.setInt(24,ent.getType());
					ps.setString(25,ent.getTypeNew());
					ps.setString(26,ent.getCategoryNew());
					ps.setString(27,ent.getStatus());
					ps.setInt(28,ent.getStatusCode());
					ps.setDate(29,ent.getStartDate());
					ps.setDate(30,ent.getTermStart());

					ps.setDate(31,ent.getTermEnd());
					ps.setDate(32,ent.getCheckDate());
					ps.setDate(33,ent.getLogoutDate());
					ps.setString(34,ent.getLogoutReason());
					ps.setDate(35,ent.getRevokeDate());
					ps.setString(36,ent.getRevokeReason());
					ps.setString(37,ent.getRevokedCertificates());
					ps.setString(38,ent.getProvinceCode());
					ps.setString(39,ent.getSource());
					ps.setTimestamp(40,ent.getCreateTime());

					ps.setTimestamp(41,ent.getRowUpdateTime());

					ps.setString(42,ent.getEid());
					ps.setString(43,ent.getEnterpriseName());
					ps.setString(44,ent.getStandardName());
					ps.setString(45,ent.getGroupName());
					ps.setString(46,ent.getGroupAbbreviation());
					ps.setString(47,ent.getLogoUrl());
					ps.setString(48,ent.getOperName());
					ps.setString(49,ent.getOperType());
					ps.setString(50,ent.getOperNameId());
					ps.setString(51,ent.getTitle());

					ps.setInt(52,ent.getTitleCode());
					ps.setBigDecimal(53,ent.getRegCapi());
					ps.setBigDecimal(54,ent.getRecCapi());
					ps.setString(55,ent.getCurrencyUnit());
					ps.setString(56,ent.getBusinessUrl());
					ps.setString(57,ent.getColleguesNum());
					ps.setString(58,ent.getBelongOrg());
					ps.setString(59,ent.getDepartment());
					ps.setString(60,ent.getScope());
					ps.setString(61,ent.getRegNo());

					ps.setString(62,ent.getCreditNo());
					ps.setString(63,ent.getOrgNo());
					ps.setInt(64,ent.getEconKindCode());
					ps.setInt(65,ent.getType());
					ps.setString(66,ent.getTypeNew());
					ps.setString(67,ent.getCategoryNew());
					ps.setString(68,ent.getStatus());
					ps.setInt(69,ent.getStatusCode());
					ps.setDate(70,ent.getStartDate());
					ps.setDate(71,ent.getTermStart());

					ps.setDate(72,ent.getTermEnd());
					ps.setDate(73,ent.getCheckDate());
					ps.setDate(74,ent.getLogoutDate());
					ps.setString(75,ent.getLogoutReason());
					ps.setDate(76,ent.getRevokeDate());
					ps.setString(77,ent.getRevokeReason());
					ps.setString(78,ent.getRevokedCertificates());
					ps.setString(79,ent.getProvinceCode());
					ps.setString(80,ent.getSource());
					ps.setTimestamp(81,ent.getCreateTime());

					ps.setTimestamp(82,ent.getRowUpdateTime());
				},
				new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
						.withUrl("jdbc:mysql://localhost:3306/db_test?useSSL=false&useUnicode=true&characterEncoding=UTF-8")
						.withDriverName("com.mysql.jdbc.Driver")
						.withUsername("flinksql")
						.withPassword("w0c53w0c53")
						.build()
			))
			.name("t_enterprise")
			.setParallelism(3);


		// execute program
		env.execute("Flink Streaming Job test");
	}
}
